package com.job.mr.sortallnum;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/*
编写reducer任务

前面两个范型参数为map任务的输出结果。
后面两个范型参数为reduce函数每次调用的输出结果
 */
public class SortNumReducer extends Reducer<IntWritable, NullWritable, IntWritable, IntWritable> {
    /*
    要实现全排序，因此只定义一个reduce任务，并且只有一个分区
    该 reduce 会拉取每个map任务输出的同一个分区的所有数据，因为只有一个分区，
    也就是该reduce 拉取了所有map的数据，
    由于reduce任务拉取完数据后会将所有的数据 使用归并排序，合并成一个有序文件，并且
    由于IntWritable的默认升序规则，因此输出一个整体升序数字的文件。
     */

//    因为只有一个reduce任务，因此可以使用成员变量，有序遍历数字的同时依次记录排名
    IntWritable numRank = new IntWritable(0);
    @Override
    protected void reduce(IntWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
//        之所以迭代values，是因为reduce 任务对具有相同key的values 组成一个集合 分组调用reduce方法
//        然而我们需要输出所有的key值，因此在迭代values 的同时，key 也会发生变化
//        但是在此场景中key是简单的数字，也可认为没有发生变化。
        for (NullWritable ignored : values) {
            //        当前数字排名加一
            numRank.set(numRank.get() + 1);
//          输出数字的排名和数字本身，默认使用制表符分割
            context.write(numRank, key);
        }
    }
}
